package rocketmq

import (
	"context"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/gogap/errors"
	"testkjcld.bhecard.com/gitlab/kit/esim/log"
	"testkjcld.bhecard.com/gitlab/kit/esim/rocketmq/internal"
)

type SubscribeOption func(*Subscriber)

func WithSubscribeTopicName(topicName string) SubscribeOption {
	return func(subscribe *Subscriber) {
		subscribe.topicName = topicName
	}
}

func WithSubscribeTransaction(transaction bool) SubscribeOption {
	return func(subscribe *Subscriber) {
		subscribe.transaction = transaction
	}
}

func WithSubscribeGroupID(groupID string) SubscribeOption {
	return func(subscribe *Subscriber) {
		subscribe.groupID = groupID
	}
}

func WithSubscribeMessageTag(messageTag string) SubscribeOption {
	return func(subscribe *Subscriber) {
		subscribe.messageTag = messageTag
	}
}

func WithSubscribeMessageNum(messageNum int) SubscribeOption {
	return func(subscribe *Subscriber) {
		subscribe.messageNum = messageNum
	}
}

func WithSubscribeMaxWait(maxWait int) SubscribeOption {
	return func(subscribe *Subscriber) {
		subscribe.maxWait = maxWait
	}
}

func WithSubscribeConcurrency(curr int) SubscribeOption {
	return func(subscribe *Subscriber) {
		subscribe.concurrency = curr
	}
}

func (se *SubscribeEngine) Subscriber(options ...SubscribeOption) error {
	sub := new(Subscriber)
	for _, opt := range options {
		opt(sub)
	}

	if sub.topicName == "" {
		return errors.Errorf("topicName[%s]非法", sub.topicName)
	}
	if sub.groupID == "" {
		return errors.Errorf("groupID[%s]非法", sub.groupID)
	}
	// 默认消息条数
	if sub.messageNum == 0 {
		sub.messageNum = 3
	}

	// 默认轮询时间
	if sub.maxWait == 0 {
		sub.maxWait = 10
	}

	// 默认并发
	if sub.concurrency == 0 {
		sub.concurrency = 1
	}

	// logger
	sub.logger = se.logger
	// 设置rocket Proxy
	sub.client = se.client
	// 处理链
	sub.handlersChain = se.combineHandlers(sub.handlersChain)

	sub.pool.New = func() interface{} {
		return sub.allocateContext()
	}

	if sub.transaction {
		se.subTransList = append(se.subTransList, sub)
		se.logger.Infof("SubscriberTrans [%+v] register success!", sub)
	} else {
		se.subList = append(se.subList, sub)
		se.logger.Infof("Subscriber [%+v] register success!", sub)
	}

	return nil
}

type Subscriber struct {
	client internal.RocketProxy

	logger log.Logger

	handlersChain HandlersChain

	pool sync.Pool

	topicName  string
	groupID    string
	messageTag string

	transaction bool

	messageNum int
	maxWait    int

	concurrency int
}

func (s *Subscriber) allocateContext() *Context {
	return &Context{subscriber: s}
}

func (s *Subscriber) handler(msg *internal.ConsumeMessage) (*internal.ConsumeMessageAck, error) {
	c := s.pool.Get().(*Context)
	defer s.pool.Put(c)

	c.reset()
	c.consumeMsg = msg
	c.consumeAck = &internal.ConsumeMessageAck{
		MessageID:     msg.Messages.MessageId,
		ReceiptHandle: "",
	}
	c.handlers = s.handlersChain

	// 调用链
	c.Next()

	if c.Errors != nil || c.IsAborted() {
		return nil, errors.Errorf("handleConsumeMsg 处理失败[%s][%v]", c.Errors, c.IsAborted())
	}

	if c.IsAck() {
		return c.consumeAck, nil
	}

	return nil, errors.Errorf("handleConsumeMsg 处理失败[%s][%v]", c.Errors, c.IsAborted())
}

func (s *Subscriber) subscribeAliyun(ctx context.Context) {
	var wg sync.WaitGroup

	respChan := make(chan internal.ConsumeMessageResponse)
	errChan := make(chan error)

	c, err := s.client.CreateConsumer(s.topicName, s.groupID, s.messageTag)
	if err != nil {
		s.logger.Errorc(ctx, "subscribeAliyun： CreateConsumer 创建失败[%s]", err)
		return
	}

	wg.Add(1)
	go func(ctx context.Context) {
		defer wg.Done()
		select {
		case <-ctx.Done():
			s.logger.Infof("ConsumeMessage 读取消息超时，返回重试；")
		case <-time.After(40 * time.Second):
			s.logger.Infof("ConsumeMessage 读取消息超时40s;")

		case resp := <-respChan: // 正常处理消息
			for i := 0; i < len(resp.Messages); i++ {
				v := internal.ConsumeMessage{Messages: resp.Messages[i]}

				s.logger.Infof("Receive MessageID: [%s], PublishTime: [%d], MessageTag: [%s] ConsumedTimes: [%d], FirstConsumeTime: [%d], NextConsumeTime: [%d]",
					v.Messages.MessageId, v.Messages.PublishTime, v.Messages.MessageTag, v.Messages.ConsumedTimes, v.Messages.FirstConsumeTime, v.Messages.NextConsumeTime)

				// 处理函数
				ack, err := s.handler(&v)
				if err != nil {
					s.logger.Errorf("handleConsumeMsg 处理失败[%s][%s]下次接收时间[%d]",
						v.Messages.MessageId, v.Messages.MessageBody, v.Messages.NextConsumeTime)
					continue
				}

				// rocketMq 发起消息确认
				ackErr := c.AckMessage([]string{ack.ReceiptHandle})
				if ackErr != nil {
					s.logger.Errorf("AckMessage[%s] [%s]确认失败[%v]", v.Messages.MessageId, ack.ReceiptHandle, ackErr)
					s.debugError(ackErr)
					continue
				}
				s.logger.Infof("Ack[%s] ---->[%s]确认成功;", v.Messages.MessageId, ack.ReceiptHandle)
			}
		case err := <-errChan: // 订阅消息失败
			if strings.Contains(err.(errors.ErrCode).Error(), "MessageNotExist") {
				s.logger.Debugf("ConsumeMessage No new message, continue")
				return
			}
			s.logger.Errorf("ConsumeMessage 读取消息失败[%s]", err)
			s.debugError(err)
		}
	}(ctx)

	// 长轮询消费消息
	// 长轮询表示如果topic没有消息则请求会在服务端挂住3s，3s内如果有消息可以消费则立即返回
	// 一次最多消费3条(最多可设置为16条)
	// 长轮询时间3秒（最多可设置为30秒）
	go func() {
		c.ConsumeMessage(respChan, errChan, int32(s.messageNum), int64(s.maxWait))
	}()

	wg.Wait()
}

func (s *Subscriber) debugError(err error) {
	if err == nil {
		return
	}
	s.logger.Errorf("阿里云RocketMQ 返回失败[%v]", err)
	/*for _, errAckItem := range err.(errors.ErrCode).Context()["Detail"].([]internal.ErrAckItem) {
		s.logger.Errorf("ErrorHandle:%s, ErrorCode:%s, ErrorMsg:%s",
			errAckItem.ErrorHandle, errAckItem.ErrorCode, errAckItem.ErrorMsg)
	}*/
}

func (s *Subscriber) subscribeApache(c internal.MQConsumer) error {
	var err error
	filter := internal.MessageSelector{}
	if s.messageTag != "" {
		filter.Type = internal.TAG
		filter.Expression = s.messageTag
	}

	s.logger.Infof("subscribeApache 订阅Apache MQ[%s][%s][%s];", s.groupID, s.topicName, s.messageTag)
	err = c.Subscribe(s.topicName, filter, func(ctx context.Context,
		msgList ...*internal.MessageExt) (internal.ConsumeResult, error,
	) {
		select {
		case <-ctx.Done():
			s.logger.Infof("apache Subscribe 退出重试；")
		default:
		}

		for _, msg := range msgList {
			s.logger.Infoc(ctx, msg.String())
			v := &internal.ConsumeMessage{
				Messages: internal.ConsumeMessageEntry{
					MessageId:     msg.MsgId,
					MessageKey:    msg.GetProperty(internal.Keys),
					MessageBody:   string(msg.Body),
					ConsumedTimes: int64(msg.ReconsumeTimes + 1),
					Properties:    msg.GetProperties(),
				},
			}

			v.Messages.StartDeliverTime, _ = strconv.ParseInt(msg.GetProperty(internal.StartDeliverTime), 10, 64)
			v.Messages.TransCheckImmunityTime, _ = strconv.Atoi(msg.GetProperty(internal.TransCheckImmunityTime))
			v.Messages.ShardingKey = msg.GetProperty(internal.SHARDING)

			//  消息处理器
			_, err = s.handler(v)
			if err != nil {
				s.logger.Errorf("handler Msg处理失败[%s][%s][%s]", v.Messages.MessageId, v.Messages.MessageBody, err)
				return internal.ConsumeRetryLater, err
			}
		}
		// 消息处理成功
		return internal.ConsumeSuccess, nil
	})
	if err != nil {
		s.logger.Errorf("subscribeApache 订阅消息[%s][%s][%s]失败[%s]", s.groupID, s.topicName, s.messageTag, err)
		return err
	}
	return nil
}
